package cn.itcast.flink.base;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author itcast
 * Date 2021/7/29 15:51
 * Desc TODO
 */
public class KeyedStateDemo {
    public static void main(String[] args) throws Exception {
        //1.env 设置并发度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.Source 参看课件
        //2.Source
        DataStreamSource<Tuple2<String, Long>> source = env.fromElements(
                Tuple2.of("北京", 1L),
                Tuple2.of("上海", 2L),
                Tuple2.of("北京", 6L),
                Tuple2.of("上海", 8L),
                Tuple2.of("北京", 3L),
                Tuple2.of("上海", 4L)
        );
        //3.Transformation
        //使用KeyState中的ValueState获取流数据中的最大值(实际中直接使用maxBy即可)
        SingleOutputStreamOperator<Tuple2<String, Long>> result1 = source.keyBy(t -> t.f0)
                //实现方式1:直接使用maxBy--开发中使用该方式即可
                .maxBy(1);
        //min只会求出最小的那个字段,其他的字段不管
        //minBy会求出最小的那个字段和对应的其他的字段
        //max只会求出最大的那个字段,其他的字段不管
        //maxBy会求出最大的那个字段和对应的其他的字段
        //实现方式2:通过managed state输入的state
        SingleOutputStreamOperator<Tuple3<String, Long, Long>> result2 = source.keyBy(t -> t.f0) /**[city,value]  -> [city,value,maxValue] */
                //3.1.先根据字符串f0分组然后进行 map 操作，将Tuple2<String/*单词*/, Long/*长度*/> 输出 Tuple3<String/*单词*/, Long/*长度*/, Long/*历史最大值*/>
                .map(new RichMapFunction<Tuple2<String, Long>, Tuple3<String, Long, Long/*maxValue*/>>() {
                    ValueState<Long> maxValueState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //存储数据的数据结构，描述
                        ValueStateDescriptor maxValueDesc = new ValueStateDescriptor("maxValue", Long.class);
                        //获取单值状态
                        maxValueState = getRuntimeContext().getState(maxValueDesc);
                    }

                    @Override
                    public Tuple3<String, Long, Long> map(Tuple2<String, Long> value) throws Exception {
                        Long historyMaxValue = maxValueState.value();
                        //当前的值
                        Long currentValue = value.f1;
                        //如果内存中的值为 0 ，或当前的值比历史内存中的值要大，把当前值存入到 内存 state 中
                        if (historyMaxValue == null || historyMaxValue < currentValue) {
                            //将当前最大值保存到内存中
                            maxValueState.update(currentValue);
                            //
                            return Tuple3.of(value.f0, currentValue, currentValue);
                        } else {
                            return Tuple3.of(value.f0, currentValue, historyMaxValue);
                        }
                    }
                });
        //-1.定义值类型的状态用来存储最大值
        //3.2.重写 RichMapFunction 的open 方法
        //-2.定义状态描述符
        //-3.从当前上下文获取内存中的状态值
        //3.3.重写 map 方法
        //-4.获取state中历史最大值value和当前元素的最大值并比较
        //-5.如果当前值大或历史值为空更新状态；返回Tuple3元祖结果
        //4.Sink 打印输出
        //result1.print();
        result2.print();
        //5.execute 执行环境
        env.execute();
    }
}
